package org.raftkv.raft.core.readonly;


import org.raftkv.raft.exceptions.RaftException;
import org.raftkv.raft.pb.Message;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

public class ReadOnly {

    private int option;

    /**
     * 使用 entry 的数据为 key，保存当前 pending 的 ReadIndex 状态
     */
    private Map<String, ReadIndexStatus> pendingReadIndex;

    /**
     * 保存 entry 的数据为队列，pending 的 ReadIndex 状态在这个队列中进行排队
     */
    private LinkedList<String> readIndexQueue;

    public ReadOnly() {
        pendingReadIndex = new HashMap<>();
        readIndexQueue = new LinkedList<>();
    }

    public ReadOnly(int option) {
        this();
        this.option = option;
    }

    public static ReadOnly newReadOnly(int option) {
        ReadOnly ro = new ReadOnly();
        ro.option = option;
        ro.pendingReadIndex = new HashMap<>();
        return ro;
    }

    /**
     * 该方法将一个只读的请求加入到 ReadOnly 中。
     *
     * @param index 是 Raft 状态机收到只读请求时的提交索引
     * @param m     是来自本地或远程节点的原始制度请求消息
     */
    public void addRequest(long index, Message m) {
        String s = new String(m.getEntries().get(0).getData());
        // 判断是否重复添加
        if (pendingReadIndex.containsKey(s)) {
            return;
        }
        pendingReadIndex.put(s, new ReadIndexStatus(index, m, new HashMap<>()));
        readIndexQueue.addLast(s);
    }

    /**
     * 该方法通知 ReadOnly Raft 状态机收到了与只读请求上下文附加的心跳的 ack。
     */
    public Map<Long, Boolean> recvAck(long id, byte[] context) {
        ReadIndexStatus rs = this.pendingReadIndex.get(new String(context));
        if (rs == null) {
            return null;
        }

        rs.getAcks().put(id, true);
        return rs.getAcks();
    }

    /**
     * 该函数推进由 ReadOnly 保存的只读请求队列。<br/>
     * 它将请求取消排队，直到找到与给定 m 具有相同上下文的只读请求。
     */
    public List<ReadIndexStatus> advance(Message m) {
        int i = 0;
        boolean found = false;

        String ctx = new String(m.getContext());
        List<ReadIndexStatus> rss = new ArrayList<>();

        for (String okctx : this.readIndexQueue) {
            i++;
            ReadIndexStatus rs = pendingReadIndex.get(okctx);
            if (rs == null) {
                throw new RaftException("cannot find corresponding read state from pending map");
            }
            // 都加入应答队列中，后面用于根据这个队列的数据来删除 pendingReadIndex 中对应的数据
            rss.add(rs);
            if (okctx.equals(ctx)) {
                found = true;
                break;
            }
        }

        if (found) {
            // 找到了，就丢弃在这之前的队列 readonly 数据了
            this.readIndexQueue = new LinkedList<>(readIndexQueue.subList(i, readIndexQueue.size()));
            rss.forEach(rs -> pendingReadIndex.remove(new String(rs.getReq().getEntries().get(0).getData())));
            return rss;
        }
        return null;
    }

    /**
     * 从 readonly 队列中返回最后一个数据
     */
    public String lastPendingRequestCtx() {
        if (readIndexQueue.size() == 0) {
            return "";
        }
        return readIndexQueue.get(readIndexQueue.size() - 1);
    }

    public int getOption() {
        return option;
    }

    public void setOption(int option) {
        this.option = option;
    }

    public LinkedList<String> getReadIndexQueue() {
        return readIndexQueue;
    }

    public void setReadIndexQueue(LinkedList<String> readIndexQueue) {
        this.readIndexQueue = readIndexQueue;
    }

    public Map<String, ReadIndexStatus> getPendingReadIndex() {
        return pendingReadIndex;
    }

    public void setPendingReadIndex(Map<String, ReadIndexStatus> pendingReadIndex) {
        this.pendingReadIndex = pendingReadIndex;
    }

}
